package com.imooc.imoocbilibili.config;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.imooc.imoocbilibili.constants.MQConstant;
import com.imooc.imoocbilibili.socket.WebSocket;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.List;

@Configuration
public class RocketMQConfig {
    @Value("${rocketmq.name.server.address}")
    private String nameServerAddr;

    @Bean("danmuProducer")
    public DefaultMQProducer danmuProducer() throws MQClientException {
        // 实例化消息生产者producer
        DefaultMQProducer producer = new DefaultMQProducer(MQConstant.DANMU_GROUP);
        // 设置NameServer的地址
        producer.setNamesrvAddr(nameServerAddr);
        // 启动producer示例
        producer.start();
        return producer;
    }

    @Bean("danmuConsumer")
    public DefaultMQPushConsumer danmuConsumer() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(MQConstant.DANMU_GROUP);
        consumer.setNamesrvAddr(nameServerAddr);
        consumer.subscribe(MQConstant.DANMU_TOPIC, "*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                MessageExt msg = msgs.get(0);
                if (msg == null) {
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                byte[] msgByte = msg.getBody();
                String bodyStr = new String(msgByte);
                JSONObject jsonObject = JSONUtil.parseObj(bodyStr);
                String key = jsonObject.getStr("key");
                String message = jsonObject.getStr("message");

                WebSocket webSocket = WebSocket.webSocketMap.get(key);
                if (webSocket.getSession().isOpen()) {
                    try {
                        webSocket.getSession().getBasicRemote().sendText(message);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        return consumer;

    }

}
